package com.me.myFramework.myRPC1;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;

/**
 * 简单的 RPC框架
 *
 * @author ME
 * @date 2020/2/24
 */
public class RpcFramework {

    private static final ThreadFactory NAMED_THREAD_FACTORY = new ThreadFactoryBuilder().setNameFormat("demo-pool-%d").build();
    private static final ExecutorService SINGLE_THREAD_POOL = new ThreadPoolExecutor(1, 1,
            0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), NAMED_THREAD_FACTORY,
            new ThreadPoolExecutor.AbortPolicy());

    /**
     * 服务提供者 暴露服务
     *
     * @param port 服务端口
     * @param serviceObj 服务实现
     */
    public static void export(int port, Object... serviceObj) throws Exception {
        if (serviceObj == null) {
            throw new IllegalArgumentException("service instance == null");
        }
        if (port <= 0 || port > 65535) {
            throw new IllegalArgumentException("Invalid port " + port);
        }
        Map<String, Object> map = new HashMap<>();
        for (Object obj : serviceObj) {
            String name = obj.getClass().getInterfaces()[0].getName();
            System.out.println("Export service [" + name + "] on port: " + port);
            map.put(name, obj);
        }

        ServerSocket server = new ServerSocket(port);
        while (true) {
            try {
                Socket socket = server.accept();
                SINGLE_THREAD_POOL.execute(new ServerHandler(socket, map));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }

    /**
     * 消费者直接引用指定服务
     *
     * @param <T> 接口泛型
     * @param interfaceClass 接口类型
     * @param host 服务器主机名
     * @param port 服务器端口
     * @return 远程服务
     */
    @SuppressWarnings("unchecked")
    public static <T> T reference(final Class<T> interfaceClass, final String host, final int port) throws Exception {
        if (interfaceClass == null) {
            throw new IllegalArgumentException("Interface class == null");
        }
        if (!interfaceClass.isInterface()) {
            throw new IllegalArgumentException("The " + interfaceClass.getName() + " must be interface class!");
        }
        if (host == null || host.length() == 0) {
            throw new IllegalArgumentException("Host == null!");
        }
        if (port <= 0 || port > 65535) {
            throw new IllegalArgumentException("Invalid port " + port);
        }
        System.out.println("Get remote service [" + interfaceClass.getName() + "] from server [" + host + ":" + port + "]");

        return (T) Proxy
            .newProxyInstance(interfaceClass.getClassLoader(), new Class<?>[]{interfaceClass}, new InvocationHandler() {

                @Override
                public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                    try (Socket socket = new Socket(host, port);
                             ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());
                             ObjectInputStream input = new ObjectInputStream(socket.getInputStream());) {
                        // 不知道为啥，不能 debug
                        output.writeUTF(interfaceClass.getName());
                        output.writeUTF(method.getName());
                        output.writeObject(method.getParameterTypes());
                        output.writeObject(args);
                        output.flush();

                        Object result = input.readObject();
                        if (result instanceof Throwable) {
                            throw (Throwable) result;
                        }

                        return result;
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }
            });
    }

    private static class ServerHandler implements Runnable {
        private final Socket socket;
        private final Map<String, Object> serviceMap;

        public ServerHandler(Socket socket, Map<String, Object> serviceMap) {
            this.socket = socket;
            this.serviceMap = serviceMap;
        }

        @Override
        public void run() {
            try (ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
                    ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream())) {
                String interfaceName = input.readUTF();
                String methodName = input.readUTF();
                Class<?>[] parameterTypes = (Class<?>[]) input.readObject();
                Object[] arguments = (Object[]) input.readObject();

                try {
                    Object serviceObj = serviceMap.get(interfaceName);
                    Method method = serviceObj.getClass().getMethod(methodName, parameterTypes);
                    Object result = method.invoke(serviceObj, arguments);
                    output.writeObject(result);
                } catch (InvocationTargetException e) {
                    output.writeObject(e.getTargetException());// 获取目标异常
                } catch (Exception e) {
                    output.writeObject(e);
                } finally {
                    output.flush();
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    socket.close();
                } catch (IOException e) {
                    System.err.println("socket close failed！");
                }
            }
        }
    }

}
